package srdb.migration.common;

import com.edb.util.PGInterval;
import org.srdbsql.util.SRmoney;
import srdb.migration.MigrationToolkit;
import srdb.migration.Utility;
import srdb.migration.i18n.ResourceHandler;
import srdb.migration.schema.metadata.Column;
import srdb.migration.schema.metadata.Table;

import java.io.*;
import java.sql.*;

public class MTKGenericBridge extends MTKBridge {
	private static final String BLOB_FILE_NAME = "mtk_blobdata";
	private static final String CLOB_FILE_NAME = "mtk_clobdata";
	private static final int BUFFER_SIZE = 1024;
	private MTKConstants.DATABASE_TYPE sourceDatabaseType;
	private MTKConstants.DATABASE_TYPE targetDatabaseType;

	public MTKGenericBridge(IMTKConnection srcConn, IMTKConnection targetConn) {
		super(srcConn, targetConn);
	}

	public MTKGenericBridge(IMTKConnection srcConn, IMTKConnection targetConn,
			MTKConstants.DATABASE_TYPE sourceDBType,
			MTKConstants.DATABASE_TYPE targetDBType) {
		super(srcConn, targetConn);
		this.sourceDatabaseType = sourceDBType;
		this.targetDatabaseType = targetDBType;
	}

	public long setTableData(Table table, int batchSize,
			MTKConstants.DATABASE_TYPE sourceDBType,
			MTKConstants.DATABASE_TYPE targetDBType) throws Exception {
		this.lastMigratedDataSize = 0L;
		long committedRowCount = 0L;
		boolean lastBatchFailure = false;
		PreparedStatement insertStmt = null;
		Statement sourceStmt = null;
		Statement destStmt = null;
		ResultSet rs = null;
		String insertStr = "";

		InputStream[] inputStreams = new InputStream[table
				.getLargeObjectsColumnCount() * batchSize];
		Reader[] readers = new Reader[table.getCLOBColumnCount() * batchSize];
		int isIndex = 0;
		int rIndex = 0;

		for (int i = 0; i < inputStreams.length; i++) {
			inputStreams[i] = null;
		}
		for (int i = 0; i < readers.length; i++) {
			readers[i] = null;
		}

		int j = 0;
		try {
			this.targetConn.getConnection().setAutoCommit(false);
			insertStr = "INSERT INTO "
					+ Utility.escape(table.getTargetSchemaName(),
							this.targetDatabaseType) + "."
					+ Utility.escape(table.getName(), this.targetDatabaseType)
					+ " VALUES (";

			for (int i = 0; i < table.getColumns().size(); i++) {
				insertStr = insertStr + "?";
				if (i < table.getColumns().size() - 1)
					insertStr = insertStr + ", ";
			}
			insertStr = insertStr + ");";
			insertStmt = this.targetConn.getConnection().prepareStatement(
					insertStr);

			destStmt = this.targetConn.getConnection().createStatement();
			destStmt.execute("TRUNCATE TABLE "
					+ Utility.escape(table.getTargetSchemaName(),
							this.targetDatabaseType) + "."
					+ Utility.escape(table.getName(), this.targetDatabaseType));

			this.targetConn.commit();

			if (table.getLargeObjectsColumnCount() > 0) {
				sourceStmt = this.srcConn.getConnection().createStatement(1003,
						1007);

				if (this.sourceDatabaseType == MTKConstants.DATABASE_TYPE.MYSQL)
					sourceStmt.setFetchSize(-2147483648);
				else {
					sourceStmt.setFetchSize(1);
				}

			} else if ((this.sourceDatabaseType == MTKConstants.DATABASE_TYPE.MYSQL)
					&& (MigrationToolkit.fetchSize == 1)) {
				sourceStmt = this.srcConn.getConnection().createStatement(1003,
						1007);

				sourceStmt.setFetchSize(-2147483648);
			} else {
				sourceStmt = this.srcConn.getConnection().createStatement();

				if (MigrationToolkit.fetchSize > 0) {
					sourceStmt.setFetchSize(MigrationToolkit.fetchSize);
				}
			}

			rs = sourceStmt.executeQuery(table
					.getSelectSQL(this.sourceDatabaseType));

			while (rs.next()) {
				for (int i = 0; i < table.getColumns().size(); i++) {
					Column column = table.getColumns().get(i);
					switch (Column.DataType.values()[column.getDataType()
							.ordinal()]) {
					case BIT:
						if ((this.sourceDatabaseType == MTKConstants.DATABASE_TYPE.SQLSERVER)
								&& ((this.targetDatabaseType == MTKConstants.DATABASE_TYPE.SRDB) || (this.targetDatabaseType == MTKConstants.DATABASE_TYPE.ENTERPRISEDB))) {
							insertStmt
									.setObject(i + 1, rs.getString(i + 1), -7);
							continue;
						}
						insertStmt.setBoolean(i + 1, rs.getBoolean(i + 1));

						break;
//					case BYTEA:
//						if ((column.isClob()) || (column.isNClob())) {
//							Clob clob = rs.getClob(i + 1);
//							if (clob != null) {
//								long clobLength = clob.length();
//								this.lastMigratedDataSize += clobLength;
//
//								readers[rIndex] = clob.getCharacterStream();
//
//								if (clobLength > 0L) {
//									insertStmt.setCharacterStream(i + 1,
//											readers[(rIndex++)],
//											(int) clobLength);
//								} else
//									insertStmt.setNull(i + 1, 2005);
//							} else {
//								insertStmt.setNull(i + 1, 2005);
//							}
//							continue;
//						}
//						insertStmt.setString(i + 1, rs.getString(i + 1));
//
//						break;
					case NUMERIC:
						if (rs.getObject(i + 1) != null) {
							insertStmt.setDouble(i + 1, rs.getDouble(i + 1));
							continue;
						}
						insertStmt.setNull(i + 1, 6);

						break;
					case FLOAT:
						// case :
						// case 10:
						Object objVal = rs.getObject(i + 1);

						if (column.isReal()) {
							if (objVal != null) {
								insertStmt.setFloat(i + 1, rs.getFloat(i + 1));
								continue;
							}
							insertStmt.setNull(i + 1, 6);
							continue;
						}
						if (column.isDouble()) {
							if (objVal != null) {
								insertStmt.setDouble(i + 1, rs.getFloat(i + 1));
								continue;
							}
							insertStmt.setNull(i + 1, 8);
							continue;
						}

						if (objVal != null) {
							insertStmt.setBigDecimal(i + 1,
									rs.getBigDecimal(i + 1));
							continue;
						}
						insertStmt.setNull(i + 1, 2);

						break;
					case TIME:
					case TIMESTAMP:
						if (column.getDataTypeString().equalsIgnoreCase(
								"DATETIMEOFFSET")) {
							insertStmt.setObject(i + 1, rs.getString(i + 1),
									1111);
							continue;
						}
						insertStmt.setTimestamp(i + 1, rs.getTimestamp(i + 1));

						break;
					case DATE:
						if (this.sourceDatabaseType == MTKConstants.DATABASE_TYPE.SQLSERVER) {
							insertStmt.setObject(i + 1, rs.getString(i + 1),
									1111);
							continue;
						}
						insertStmt.setTime(i + 1, rs.getTime(i + 1));

						break;
					case BYTEA:
						if (column.isBlob()) {
							Blob blob = rs.getBlob(i + 1);
							if (blob != null) {
								long blobLength = blob.length();
								this.lastMigratedDataSize += blobLength;

								inputStreams[isIndex] = rs
										.getBinaryStream(i + 1);
								if (blobLength >= 0L) {
									insertStmt.setBinaryStream(i + 1,
											inputStreams[(isIndex++)],
											(int) blobLength);
								} else
									insertStmt.setNull(i + 1, -2);
							} else {
								insertStmt.setNull(i + 1, -2);
							}

							continue;
						}
						byte[] b = rs.getBytes(i + 1);
						if (b != null) {
							this.lastMigratedDataSize += b.length;
							inputStreams[isIndex] = new ByteArrayInputStream(b);

							insertStmt.setBinaryStream(i + 1,
									inputStreams[(isIndex++)], b.length);
						} else {
							insertStmt.setNull(i + 1, -2);
						}

						break;
					case INTERVAL:
						PGInterval pgInterval = new PGInterval();
						String intervalValue = rs.getString(i + 1);

						if (intervalValue != null) {
							intervalValue = convertToEDBIntervalFormat(intervalValue);
							pgInterval.setValue(intervalValue);
							this.lastMigratedDataSize += intervalValue
									.getBytes().length;
						}

						insertStmt.setObject(i + 1, pgInterval);

						break;
					case GEOMETRIC:
						// case :
						// case :
						insertStmt.setObject(i + 1, rs.getString(i + 1), 1111);
						break;
					case MONEY:
						if (rs.getString(i + 1) != null) {
							if (MigrationToolkit.getInstance()
									.getTargetDBType() == MTKConstants.DATABASE_TYPE.SRDB) {
								insertStmt.setObject(i + 1,
										new SRmoney(rs.getString(i + 1)));
								continue;
							}
							insertStmt.setObject(i + 1, rs.getString(i + 1),
									1111);
							continue;
						}

						insertStmt.setNull(i + 1, 0);

						break;
					// case 20:
					// if ((this.sourceDatabaseType !=
					// MTKConstants.DATABASE_TYPE.SQLSERVER)
					// || (!column.getDataTypeString()
					// .equalsIgnoreCase("SQL_VARIANT")))
					// break;
					// insertStmt.setObject(i + 1, rs.getString(i + 1), 1111);
					// break;
					// }

					// if ((this.sourceDatabaseType ==
					// MTKConstants.DATABASE_TYPE.MYSQL)
					// && (table.isEnum(column.getDataTypeString())))
					// insertStmt.setObject(i + 1, rs.getString(i + 1), 1111);
					// else {
					// throw new Exception("Invalid DataType "
					// + column.getDataTypeString());
					// }
					}

					insertStmt.addBatch();
					j++;

					if (j % batchSize != 0)
						continue;
					try {
						insertStmt.executeBatch();
						lastBatchFailure = false;
						Utility.getLogger().info(
								"["
										+ table.getName()
										+ "] "
										+ ResourceHandler.getValue(
												"INFO_MIGRATED_ROWS",
												new Object[] { Integer
														.valueOf(j) }));
					} catch (SQLException ex) {
						if (ex.getNextException() != null) {
							Utility.getLogger().warn(
									ex.getNextException().getMessage()
											+ ", Skipping Batch");
							Utility.getLogger().error("Stack Trace:",
									ex.getNextException());
						} else {
							Utility.getLogger().warn(
									ex.getMessage() + ", Skipping Batch");
							Utility.getLogger().error("Stack Trace:", ex);
						}
						MigrationToolkit.incrementFailureCount();
						table.incrementBatchFailureCount();
						lastBatchFailure = true;
					} catch (Exception ex) {
						Utility.getLogger().warn(
								ex.getMessage() + ", Skipping Batch");
						Utility.getLogger().error("Stack Trace:", ex);
						MigrationToolkit.incrementFailureCount();
						table.incrementBatchFailureCount();
						lastBatchFailure = true;
					}

					insertStmt.clearBatch();
					this.targetConn.commit();

					if (!lastBatchFailure) {
						committedRowCount += batchSize;
					}

					for (int m = 0; m < inputStreams.length; m++) {
						if (inputStreams[m] != null) {
							inputStreams[m].close();
							inputStreams[m] = null;
							new File(Utility.getTempPath() + "mtk_blobdata"
									+ "_" + table.getFullName() + m).delete();
						}
					}
					for (int n = 0; n < readers.length; n++) {
						if (readers[n] != null) {
							readers[n].close();
							readers[n] = null;
							new File(Utility.getTempPath() + "mtk_clobdata"
									+ "_" + table.getFullName() + n).delete();
						}
					}
					isIndex = 0;
					rIndex = 0;
				}

				int[] lastBatchCmdCount = null;
				try {
					lastBatchCmdCount = insertStmt.executeBatch();
					lastBatchFailure = false;

					if (lastBatchCmdCount.length > 0) {
						committedRowCount += j % batchSize;
						Utility.getLogger()
								.info("["
										+ table.getName()
										+ "] "
										+ ResourceHandler
												.getValue(
														"INFO_MIGRATED_ROWS",
														new Object[] { Long
																.valueOf(committedRowCount) }));
					}
				} catch (Exception ex) {
					Utility.getLogger().warn(
							ex.getMessage() + ", Skipping Batch");
					Utility.getLogger().error("Stack Trace:", ex);
					MigrationToolkit.incrementFailureCount();
					table.incrementBatchFailureCount();
					lastBatchFailure = true;
				}
				insertStmt.clearBatch();
				this.targetConn.commit();
			}
		} catch (SQLException sqle) {
			this.targetConn.rollback();
			throw Utility
					.processException(sqle.getNextException() != null ? sqle
							.getNextException() : sqle);
		} catch (Exception ex) {
			this.targetConn.rollback();
			throw ex;
		} finally {
			for (int i = 0; i < inputStreams.length; i++) {
				if (inputStreams[i] != null) {
					inputStreams[i].close();
					inputStreams[i] = null;
					new File(Utility.getTempPath() + "mtk_blobdata" + "_"
							+ table.getFullName() + i).delete();
				}
			}
			for (int i = 0; i < readers.length; i++) {
				if (readers[i] != null) {
					readers[i].close();
					readers[i] = null;
					new File(Utility.getTempPath() + "mtk_clobdata" + "_"
							+ table.getFullName() + i).delete();
				}
			}

			this.targetConn.getConnection().setAutoCommit(true);
			if (insertStmt != null) {
				insertStmt.close();
			}
			if (destStmt != null) {
				destStmt.close();
			}
			if (rs != null) {
				rs.close();
			}
			if (sourceStmt != null) {
				sourceStmt.close();
			}
		}
		return committedRowCount;
	}

	public long setTableDataInSafeMode(Table table,
			MTKConstants.DATABASE_TYPE sourceDBType,
			MTKConstants.DATABASE_TYPE targetDBType) throws Exception {
		return -1L;
	}

	private File readLOBIntoFile(InputStream in, String tableName, int colIndex)
			throws Exception {
		byte[] buffer = new byte[1024];

		File lobFile = new File(Utility.getTempPath() + "mtk_blobdata" + "_"
				+ tableName + colIndex);
		FileOutputStream out = new FileOutputStream(lobFile);
		int length = -1;
		while ((length = in.read(buffer)) != -1) {
			out.write(buffer, 0, length);
		}
		in.close();
		out.close();

		return lobFile;
	}

	private String convertToEDBIntervalFormat(String intervalValue) {
		if (intervalValue.indexOf('-') >= 0) {
			String[] yearMonthParts = intervalValue.split("-");
			String edbIntervalValue = null;

			if (yearMonthParts.length == 2) {
				edbIntervalValue = yearMonthParts[0] + " years "
						+ yearMonthParts[1] + " months";
			}

			return edbIntervalValue;
		}

		return intervalValue;
	}
}

/*

 */